package com.wh.stressservice.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.msf.modules.biz.entity.StDevice;
import com.msf.modules.biz.entity.StSensor;
import com.msf.modules.biz.service.IStDeviceService;
import com.msf.modules.biz.service.IStSensorService;
import com.wh.influxdb.measurement.Stress;
import com.wh.influxdb.service.IStressService;
import com.wh.stressservice.entity.Device;
import com.wh.stressservice.service.IAccessService;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.annotations.Select;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author xu
 * @Date 2025/1/21
 **/
@Slf4j
@Service
public class AccessServiceImpl implements IAccessService {

    private final IStressService stressService;
    private final JdbcTemplate accessJdbcTemplate;
    private final IStDeviceService stDeviceService;
    private final IStSensorService stSensorService;

    private static final int BATCH_SIZE = 10000;
    private static final int MAX_RETRY_TIMES = 2;
    private final ExecutorService executorService;

    public AccessServiceImpl(
            IStressService stressService,
            @Qualifier("accessJdbcTemplate") JdbcTemplate accessJdbcTemplate,
            IStDeviceService stDeviceService,
            IStSensorService stSensorService, ExecutorService executorService) {
        this.stressService = stressService;
        this.accessJdbcTemplate = accessJdbcTemplate;
        this.stDeviceService = stDeviceService;
        this.stSensorService = stSensorService;
        // 创建线程池，核心线程数为CPU核心数
        this.executorService = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors(),
                new ThreadFactoryBuilder().setNameFormat("stress-sync-pool-%d").build());
    }

    @Override
    // @Async
    public void syncData() {
        // this.syncDevice();
        // this.syncSensor();
        this.realTimeData();
    }

    @Override
    public void syncDevice() {
        String sql = "select MID AS id,MName AS mName,MType AS mType,MStatus AS mStatus from mcu ";
        log.info("**********开始查询数据********");
        List<Map<String, Object>> obj = accessJdbcTemplate.queryForList(sql);

        log.info("**********数据查询成功********");
        List<Device> devices = Collections.emptyList();
        if (!obj.isEmpty()) {
            devices = obj.stream().map(map -> {
                Device device = new Device();
                device.setId(String.valueOf(map.get("id")));
                device.setMName(String.valueOf(map.get("mName")));
                device.setMType(String.valueOf(map.get("mType")));
                device.setMStatus(String.valueOf(map.get("mStatus")));
                return device;
            }).collect(Collectors.toList());
        }
        if (!devices.isEmpty()) {
            for (Device device : devices) {
                StDevice stDevice = stDeviceService.getById(device.getId());
                if (stDevice == null) {
                    stDevice = new StDevice();
                    stDevice.setId(device.getId());
                    stDevice.setMName(device.getMName());
                    stDevice.setCreateTime(new Date(System.currentTimeMillis()));
                    stDeviceService.save(stDevice);
                }
                stDevice.setMName(device.getMName());
                stDevice.setMType(device.getMType());
                stDevice.setMStatus(device.getMStatus());
                stDevice.setUpdateTime(new Date(System.currentTimeMillis()));
                stDeviceService.updateById(stDevice);
            }
            log.info("应力设备同步成功，总共{}条数据", devices.size());
        }
    }

    @Override
    public void syncSensor() {
        String sql = "select SID,MID,SName,SensorType AS SType,ST1,ST2,Unit1,Unit2 from sensor ";
        List<Map<String, Object>> obj = accessJdbcTemplate.queryForList(sql);
        List<StSensor> sensors = Collections.emptyList();
        if (!obj.isEmpty()) {
            sensors = obj.stream().map(map -> {
                StSensor stSensor = new StSensor();
                stSensor.setId(String.valueOf(map.get("SID")));
                stSensor.setMId(String.valueOf(map.get("MID")));
                stSensor.setSName(String.valueOf(map.get("SName")));
                stSensor.setSType(String.valueOf(map.get("SType")));
                stSensor.setST1(String.valueOf(map.get("ST1")));
                stSensor.setST2(String.valueOf(map.get("ST2")));
                stSensor.setUnit1(String.valueOf(map.get("Unit1")));
                stSensor.setUnit2(String.valueOf(map.get("Unit2")));
                return stSensor;
            }).collect(Collectors.toList());
        }
        if (!sensors.isEmpty()) {
            for (StSensor sensor : sensors) {
                StSensor stSensor = stSensorService.getById(sensor.getId());
                if (stSensor == null) {
                    sensor.setCreateTime(new Date(System.currentTimeMillis()));
                    stSensorService.save(sensor);
                }
                StDevice stDevice = stDeviceService.getById(sensor.getMId());
                sensor.setSStatus(stDevice.getMStatus());
                sensor.setUpdateTime(new Date(System.currentTimeMillis()));
                stSensorService.updateById(sensor);
            }
            log.info("应力传感器同步成功，总共{}条数据", sensors.size());
        }
    }

    @Override
    public void realTimeData() {
        /**
         * @Description:
         *               1、查询influxdb时序数据库中Measurement为stress的最新一条数据
         *               2、如果1中最新的数据为空同步access数据库中DATA表中的所有数据并出入到influxdb时序数据库中Measurement为stress里面并且数据需需要保存两年。
         *               因为access数据库中DATA表中数据量非常大，请考虑全表同步时的性能为题及可行性。
         *               3、如果1中最新的数据不为空，根据最新数据的时间戳做增量同步，从access数据库中DATA表中同步增量数据到时序数据库中Measurement为stress里面并且数据需需要保存两年。
         **/
        try {
//            Stress stress = stressService.queryLastRow();
//            if (stress == null) {
//                handleFullSync();
//            } else {
//                handleIncrementalSync(Date.from(stress.getTime()));
//            }
            handleFullSync();
        } catch (Exception e) {
            log.error("数据同步过程发生错误", e);
        }
    }

    private void handleFullSync() {
        log.info("开始全量同步数据...");
        try {
            // 获取总记录数
            String countSql = "SELECT COUNT(*) FROM data";
            int totalCount = accessJdbcTemplate.queryForObject(countSql, Integer.class);
            int totalPages = (totalCount + BATCH_SIZE - 1) / BATCH_SIZE;

            // 创建任务列表
            List<CompletableFuture<Void>> futures = new ArrayList<>();
            for (int page = 0; page < totalPages; page++) {
                final int currentPage = page;
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    processBatch(currentPage, null);
                }, executorService).exceptionally(throwable -> {
                    log.error("批次{}处理失败", currentPage, throwable);
                    return null;
                });
                futures.add(future);
            }

            // 等待所有任务完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            log.info("全量同步完成，共处理{}条数据", totalCount);
        } catch (Exception e) {
            log.error("全量同步过程发生错误", e);
            throw new RuntimeException("全量同步失败", e);
        }
    }

    private void handleIncrementalSync(Date lastSyncTime) {
        log.info("开始增量同步数据，同步时间点：{}", lastSyncTime);
        try {
            processBatch(0, lastSyncTime);
        } catch (Exception e) {
            log.error("增量同步过程发生错误", e);
            throw new RuntimeException("增量同步失败", e);
        }
    }

    private void processBatch(int page, Date lastSyncTime) {
        int retryCount = 0;
        while (retryCount < MAX_RETRY_TIMES) {
            try {
                StringBuilder sqlBuilder = new StringBuilder();
                sqlBuilder.append("SELECT * FROM ( ");
                sqlBuilder.append("    SELECT TOP ? * FROM ( ");
                sqlBuilder.append("        SELECT TOP ? d.S1, d.S2, d.R1, d.R2, d.SID, s.SName, d.DataTime ");
                sqlBuilder.append("        FROM data d ");
                sqlBuilder.append("        LEFT JOIN sensor s ON d.SID = s.SID ");
                if (lastSyncTime != null) {
                    sqlBuilder.append(" WHERE d.DataTime > ? ");
                }
                sqlBuilder.append("        ORDER BY d.DataTime ASC");
                sqlBuilder.append("    ) AS T1 ");
                sqlBuilder.append("    ORDER BY DataTime DESC");
                sqlBuilder.append(") AS T2 ");
                sqlBuilder.append("ORDER BY DataTime ASC");

                List<Map<String, Object>> dataList;
                if (lastSyncTime != null) {
                    dataList = accessJdbcTemplate.queryForList(
                            sqlBuilder.toString(),
                            lastSyncTime,
                            page * BATCH_SIZE,
                            BATCH_SIZE);
                } else {
                    dataList = accessJdbcTemplate.queryForList(
                            sqlBuilder.toString(),
                            BATCH_SIZE,
                            page * BATCH_SIZE);
                }

                if (!dataList.isEmpty()) {
                    List<Stress> stressList = convertToStressList(dataList);
                    saveWithRetry(stressList);
                    log.info("第{}批次数据同步完成，本次同步{}条数据", page, stressList.size());
                }
                break; // 成功后跳出重试循环
            } catch (Exception e) {
                retryCount++;
                if (retryCount >= MAX_RETRY_TIMES) {
                    log.error("批次{}处理失败，已重试{}次", page, retryCount, e);
                    throw new RuntimeException("批次处理失败", e);
                }
                log.warn("批次{}处理失败，准备第{}次重试", page, retryCount + 1);
                try {
                    Thread.sleep(1000 * retryCount); // 重试延迟
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    private List<Stress> convertToStressList(List<Map<String, Object>> dataList) {
        return dataList.stream().map(map -> {
                    try {
                        Stress newStress = new Stress();
                        newStress.setTime(((Date) map.get("DataTime")).toInstant());
                        newStress.setS1(parseDoubleValue(map.get("S1")));
                        newStress.setS2(parseDoubleValue(map.get("S2")));
                        newStress.setR1(parseDoubleValue(map.get("R1")));
                        newStress.setR2(parseDoubleValue(map.get("R2")));
                        newStress.setSName(String.valueOf(map.get("SName")));
                        return newStress;
                    } catch (Exception e) {
                        log.error("数据转换错误: {}", map, e);
                        return null;
                    }
                }).filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    private double parseDoubleValue(Object value) {
        if (value == null || ObjectUtil.isEmpty(value))
            return 0.0;
        try {
            return Double.parseDouble(String.valueOf(value));
        } catch (NumberFormatException e) {
            log.warn("数值转换失败: {}", value);
            return 0.0;
        }
    }

    private void saveWithRetry(List<Stress> stressList) {
        int retryCount = 0;
        while (retryCount < MAX_RETRY_TIMES) {
            try {
                stressService.batchInsert(stressList);
                break;
            } catch (Exception e) {
                retryCount++;
                if (retryCount >= MAX_RETRY_TIMES) {
                    throw new RuntimeException("保存数据失败", e);
                }
                log.warn("保存数据失败，准备第{}次重试", retryCount + 1);
                try {
                    Thread.sleep(1000 * retryCount);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    @PreDestroy
    public void destroy() {
        if (executorService != null) {
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

}